package ds_industry_2025.ds.ds_03.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties
/*
    4、根据dws层表来计算每个地区2020年订单金额前3省份，依次存入MySQL数据库shtd_result的regiontopthree表中（表结构如下），然后
    在Linux的MySQL命令行中根据地区表主键升序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对
    应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下；
 */
object t4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t4")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    spark.table("dws.province_consumption_day_aggr")
      .where(col("year")=== 2020)
      .createOrReplaceTempView("data")

    val r1 = spark.sql(
      """
        |select
        |region_id,
        |region_name,
        |province_id,
        |province_name,
        |total_amount,
        |row
        |from(
        |select distinct
        |region_id,
        |region_name,
        |province_id,
        |province_name,
        |total_amount,
        |row_number() over(partition by region_id,region_name order by total_amount desc) as row
        |from data
        |) as r1
        |where row < 4
        |""".stripMargin)

    val conn = new Properties()
    conn.setProperty("user", "root")
    conn.setProperty("password", "123456")
    conn.setProperty("driver", "com.mysql.jdbc.Driver")



    //  todo 分组后不能使用withColumn去添加列，只能使用聚合函数
    //  todo 而且collect_list是一个聚合函数不能和withColumn一起使用
    //  todo groupBy然后agg聚合之后只会保留分组字段和聚合字段，所以，如果需要聚合多列就得在一个agg里面完成，否则找不到字段报错
    //  todo concat_ws(",",col())   concat_ws将数据转化为字符串，以","号为分隔符
    //  todo 同样的，也可以在sql里面使用concat_ws(",",collect_list("province_id"))
    r1.groupBy("region_id","region_name")
      .agg(
        concat_ws(",",collect_list("province_id")).as("provinceids"),
        concat_ws(",",collect_list("province_name")).as("provincenames"),
        concat_ws(",",collect_list("total_amount")).as("provinceamount")
      )
      .write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_result?useSSL=false","regiontopthree",conn)







    spark.close()
  }

}
